package com.packtpub.esh;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
*  执行的时候可以将单词和对应的汇总结果推送到Elasticsearch的索引当中
*  {
*          "count": 4,
*          "word": "Apache"
*  }
*  我们需要在Java中实现这样的一个文档结构。它应该被写入到Context对象中
*  将这个基于JSON的键值对映射到MapWritable类上
*  MapWritable中的键对应JSON中的键， MapWritable 中的值对应JSON中的值。
*  WordsReducer实现了Reducer接口，以＜Text, Iterable<IntWritable>作为输入参数类型，以<Text, MapWritable>作为输出参数类型
*/
public class WordsReducer extends Reducer<Text,IntWritable,Text,MapWritable> {

	@Override
	public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
		// This represents our ES document
        MapWritable result = new MapWritable();
		//们对对应单词的频度进行迭代累加，得到了总的出现次数
		int sum = 0;
		for (IntWritable val : values) {
			sum += val.get();
		}
		//总的出现次数和对应的单词被存入Map Writable对象result
        // Add "word" field to ES document
		result.put(new Text("word"), key);
        // Add "count" field to ES document
		result.put(new Text("count"), new IntWritable(sum));
		//最后， result被写入context进行进一步处理
		context.write(key, result);
	}
	
}
